package com.stillalive.mq.mqserver;

import com.stillalive.mq.common.Consumer;
import com.stillalive.mq.common.MqException;
import com.stillalive.mq.mqserver.core.*;
import com.stillalive.mq.mqserver.datacenter.DiskDataCenter;
import com.stillalive.mq.mqserver.datacenter.MemoryDataCenter;
import lombok.Getter;
import lombok.Setter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 通过这个类来表示虚拟主机
 * 每个虚拟主机下面都管理着自己的交换机，队列，绑定，消息数据
 * 同时提供API供上层调用
 * 作为业务逻辑的整合者，需要对代码中抛出的异常进行处理
 */
@Getter
@Setter
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new  MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new  DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    // 操作交换机的锁对象
    private final Object exchangeLocker = new Object();
    // 操作队列的锁对象
    private final Object queueLocker = new Object();


    public VirtualHost(String name) {
        this.virtualHostName = name;
        // MemoryDataCenter不需要额外的初始化操作，直接new出来对象即可
        // DiskDataCenter则需要调用init方法进行建库建表和初始数据的设定
        diskDataCenter.init();
        // 还需要将硬盘数据恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败！");
        }
    }

    // 创建交换机
    // 不存在就创建，存在则直接返回
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) {
        // 交换机名字加上虚拟主机作为前缀
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker) {
                // 判断交换机是否已经存在
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                // 已经存在
                if(existsExchange != null){
                    System.out.println("[VirtualHost] 交换机已经存在！exchangeName=" +  exchangeName);
                    return true;
                }
                // 不存在，构建一个
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                exchange.setType(exchangeType);
                // 交换机对象写入硬盘（需要持久化的才写）
                if(durable){
                    diskDataCenter.insertExchange(exchange);
                }
                // 交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建成功！exchangeName=" +  exchangeName);
                // 上述逻辑，先写硬盘，后写内存，是因为写硬盘操作更容易失败，如果硬盘写入失败，内存就不写了
                // 反之写完内存还需要删掉
                return true;
            }
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机创建失败！exchangeName=" +  exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除交换机
    public boolean exchangeDelete(String  exchangeName) {
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker) {
                // 找到对应的交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if(toDelete == null){
                    throw new MqException("[VirtualHost] 交换机不存在无法删除！exchangeName=" +  exchangeName);
                }
                // 删除硬盘上的数据
                if(toDelete.isDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                // 删除内存中的交换机数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功！exchangeName=" +  exchangeName);
                return true;
            }
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机删除失败！exchangeName=" +  exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 找到对应的交换机
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在！queueName=" +  queueName);
                    return true;
                }
                // 创建队列对象
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                // 写硬盘
                if(durable){
                    diskDataCenter.insertQueue(queue);
                }
                // 写内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功！queueName=" +  queueName);
                return true;
            }
        }catch(Exception e){
            System.out.println("[VirtualHost] 队列创建失败！queueName=" +  queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除队列
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try{
            synchronized (queueLocker) {
                // 根据队列名字查询队列对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null){
                    throw new MqException("[VirtualHost] 队列不存在无法删除！queueName=" +  queueName);
                }
                // 删除硬盘中的数据
                if(queue.isDurable()){
                    diskDataCenter.deleteQueue(queueName);
                }
                // 删除内存中的数据
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除成功！queueName=" +  queueName);
                return true;
            }
        }catch(Exception e){
            System.out.println("[VirtualHost] 队列删除失败！queueName=" +  queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 创建绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 判定当前绑定是否已经存在
                    Binding existsBinding = memoryDataCenter.getBinding(queueName, exchangeName);
                    if(existsBinding != null){
                        throw new MqException("[VirtualHost] binding已经存在！queueName=" + queueName
                                + "，exchangeName=" + exchangeName);
                    }
                    // 验证bindingKey是否合法
                    if(!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] bindingKey非法！bindingKey=" + bindingKey);
                    }
                    // 创建Binding对象
                    Binding binding = new Binding();
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    binding.setBindingKey(bindingKey);
                    // 获取对应的交换机和队列，如果交换机或队列不存在，同样无法创建抛异常
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if(queue == null){
                        throw new MqException("[VirtualHost] 队列不存在！queueName=" + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(exchange == null){
                        throw new MqException("[VirtualHost] 交换机不存在！exchangeName=" + exchangeName);
                    }
                    // 先写硬盘
                    if(queue.isDurable() && exchange.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }
                    // 再写内存
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 绑定创建成功! exchangeName=" + exchangeName
                            + ", queueName=" + queueName);
                    return true;
                }
            }
        }catch (Exception e){
            System.out.println("[VirtualHost] 绑定创建失败! exchangeName=" + exchangeName
                    + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }
    public boolean queueUnbind(String queueName, String exchangeName){
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 获取 binding 看是否已经存在
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MqException("[VirtualHost] 删除绑定失败! 绑定不存在! exchangeName=" + exchangeName + ", queueName=" + queueName);
                    }
                    // 2. 无论绑定是否持久化了, 都尝试从硬盘删一下. 就算不存在, 这个删除也无副作用
                    diskDataCenter.deleteBinding(binding);
                    // 3. 删除内存的数据
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 删除绑定成功!");
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除绑定失败!");
            e.printStackTrace();
            return false;
        }
    }

    // 发送消息到指定队列中
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        exchangeName = virtualHostName + exchangeName;
        try{
            // 检查routingKey是否合法
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] routingKey非法！routingKey=" + routingKey);
            }
            // 查找交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null){
                throw new MqException("[VirtualHost] 交换机不存在！exchangeName=" + exchangeName);
            }
            // 判断交换机类型
            if(exchange.getType() == ExchangeType.DIRECT){
                // 按照直接交换机的方式来转发消息
                // 以routingKey作为队列名字，直接转发给队列，此时可以无视绑定关系
                String queueName = virtualHostName + routingKey;
                // 构造消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                // 查找队列名对应的对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null){
                    throw new MqException("[VirtualHost] 队列不存在！queueName=" + queueName);
                }
                sendMessage(queue, message);
            }else {
                // 按照 FANOUT 和 TOPIC 的方式来转发
                // 找到交换机关联的所有绑定，并遍历这些绑定对象
                ConcurrentHashMap<String, Binding> bindingMap = memoryDataCenter.getBindings(exchangeName);
                for(Map.Entry<String, Binding> entry : bindingMap.entrySet()){
                    // 获取到绑定对象，判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue == null){
                        System.out.println("[VirtualHost] basicPublish发送消息时，发现队列不存在！binding=" + binding);
                        continue;
                    }
                    // 构造消息对象
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // 判定这个消息是否能转发给该队列
                    // fanout是所有的绑定队列都要转发
                    // topic需要判定bindingKey和routingKey是否匹配
                    if(!router.route(exchange.getType(), binding, message)){
                        continue;
                    }
                    // 真正转发给队列
                    sendMessage(queue, message);
                }
            }
            return true;
        }catch(Exception e){
            System.out.println("[VirtualHost] 消息发送失败！");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        // 把消息按需写入到硬盘和内存上
        int deliverMode = message.getDeliverMode();
        // deliverMode 为1，不持久化；为2，持久化
        if(deliverMode == 2){
            diskDataCenter.sendMessage(queue, message);
        }
        memoryDataCenter.sendMessage(queue, message);
        // 通知消费者可以消费消息
        consumerManager.notifyConsume(queue.getName());
    }

    // 订阅消息
    // 添加一个队列的订阅者，当队列收到消息后，就要把消息推送给对应的订阅者
    // consumer 是一个回调函数，定义为函数式接口
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        // 构造一个ConsumerEnv对象，把对应的队列找到，再把Consumer对象添加到队列中
        queueName = virtualHostName + queueName;
        try{
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume 成功！queueName=" + queueName);
            return true;
        }catch(Exception e){
            System.out.println("[VirtualHost] basicConsume 失败！queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }
    public boolean basicAck(String queueName, String messageId){
        queueName = virtualHostName + queueName;
        try{
            // 获取到消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if(message == null){
                throw new MqException("[VirtualHost] 要确认的消息不存在！messageId=" + messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null){
                throw new MqException("[VirtualHost] 要确认的队列不存在！queueName=" + queueName);
            }
            // 删除硬盘上的数据
            if(message.getDeliverMode() == 2){
                diskDataCenter.deleteMessage(queue, message);
            }
            // 删除消息中心的数据
            memoryDataCenter.removeMessage(messageId);
            // 删除待确认集合中的数据
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] basicAck成功！消息确认成功！queueName=" + queueName
                        + ", messageId=" + messageId);
            return true;
        }catch(Exception e){
            System.out.println("[VirtualHost] basicAck成功！消息确认失败！queueName=" + queueName
                    + ", messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }
}
